package com.gugee.tiktok.data.consumer.fansportrait.core.main;


import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.tiktok.common.model.constant.ConstAi;
import com.gugee.tiktok.common.model.dal.blogger.AwemeCommentRepository;
import com.gugee.tiktok.common.model.dal.blogger.AwemeRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerFansPortraitRepository;
import com.gugee.tiktok.common.model.dal.blogger.BloggerRepository;
import com.gugee.tiktok.common.model.dto.SlimAwemeCommentDto;
import com.gugee.tiktok.common.model.dto.SlimBloggerAiExtDto;
import com.gugee.tiktok.common.model.mysql.Blogger;
import com.gugee.tiktok.common.model.mysql.BloggerFansPortrait;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.common.msg.FansPortraitMessage;
import com.gugee.tiktok.data.consumer.fansportrait.config.FansPortraitAppConfig;
import com.xiguaji.boot.toolkit.DateTimeUtils;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;

/**
 * Description:
 * User: Huang xin
 * Date: 2020-11-12 18:06
 */
@Component
@Slf4j
public class SaveFansPortraitMain {

    @Autowired
    ScheduleTemplate scheduleTemplate;

    @Autowired
    RestHighLevelClient highLevelClient;
    private String workName = "fansPortrait-worker";

    @Autowired
    FansPortraitAppConfig fansPortraitAppConfig;

    @Autowired
    KafkaTemplate<Long, FansPortraitMessage> kafkaTemplateFansPortrait;

    @Autowired
    BloggerRepository bloggerRepository;

    @Autowired
    BloggerFansPortraitRepository bloggerFansPortraitRepository;

    @Autowired
    AwemeCommentRepository awemeCommentRepository;

    @Autowired
    SlimAwemeCommentDto.SlimAwemeCommentMapper slimAwemeCommentMapper;

    @Autowired
    KafkaTemplate<Long, BloggerMessage> kafkaTemplateBlogger;

    @Autowired
    AwemeRepository awemeRepository;

    ForkJoinPool forkJoinPool;

    /**
     * redis服务
     */
    @Autowired
    RedisTemplate<String,String> redisTemplate;

    /**
     * 用来转换json string到实体类
     */
    final ObjectMapper objectMapper = new ObjectMapper();

    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(fansPortraitAppConfig.getFansPortraitPollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     *
     * @param msgs 粉丝画像消息体
     */
    public void handle(List<FansPortraitMessage> msgs)   {
        try {
            forkJoinPool.submit(() -> msgs.parallelStream().forEach(msg -> {
                try {
                    //开始分析粉丝画像
                    runStatisticsFansPortrait(msg);
                } catch (Exception ex) {
                    log.error("FansPortrait error. awemeId:{},{}" , msg.getAwemeId(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("FansPortrait thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 重试
     * @param msg
     * @param awemeId
     * @param retryCause
     */
    private void retryMessage(FansPortraitMessage msg,long awemeId,String retryCause){
        log.error("{},awemeId={},将投递重试={} ",retryCause, awemeId,  msg.getRetryTimes());
        if(msg.getRetryTimes()>0){
            msg.setRetryTimes(msg.getRetryTimes()-1);
            kafkaTemplateFansPortrait.send(fansPortraitAppConfig.getFansPortraitTopicOnline(),msg);
        }else{
            log.error("重试{}次后依旧失败. awemeId:{}" ,msg.getRetryLimitTimes(), msg.getAwemeId());
        }
    }

    /**
     * 开始分析粉丝画像
     * @param fansPortraitMessage
     */
    private void runStatisticsFansPortrait(FansPortraitMessage fansPortraitMessage) {
        long awemeId = fansPortraitMessage.getAwemeId();
        long uid = fansPortraitMessage.getUid();
        StringBuffer eventName = new StringBuffer("SaveAwemeFansPortraitEvent");

        List<SlimAwemeCommentDto> awemeComments = fansPortraitMessage.getComments();

        if(fansPortraitMessage.getId() == null || fansPortraitMessage.getId() == 0){
            log.error("分析粉丝画像-接收到的消息无id值，awemeId={}",awemeId);
            return;
        }

        if(awemeComments == null ||awemeComments.size() == 0){
            log.debug("分析粉丝画像-消息中无评论，尝试从oss中读取，awemeId={}",awemeId);
            eventName.append(" FromOss ");
            String errorMsg = "oss中无评论";
            try{
                //根据视频id从oss获取评论
                //awemeComments = findCommentByAwemeId(awemeId);
                val aweme = awemeRepository.findByoAwemeId(awemeId);
                int dateCode = 0;
                if(aweme.isPresent()){
                    dateCode = aweme.get().getDatecode();
                }
                awemeComments = objectMapper.readValue(awemeCommentRepository.get(awemeId,dateCode), new TypeReference<List<SlimAwemeCommentDto>>(){});
            }catch (Exception e){
                e.printStackTrace();
                errorMsg = e.getMessage();
            }
            if(awemeComments == null ||awemeComments.size() == 0){
                log.error("oss中读取评论失败，awemeId={}，异常：{}",awemeId,errorMsg);
                return;
            }
        }else{
            eventName.append(" FromCommentCollect ");
        }

        //分析保存
        statisticsAwemeFansPortrait(uid,awemeId,awemeComments,eventName.toString());
    }

    /**
     * 解析视频的粉丝画像信息
     * @param uid 博主id
     * @param awemeId 视频id
     * @param awemeComments 视频的评论
     * @param eventName 事件名称
     */
    private void statisticsAwemeFansPortrait(long uid,long awemeId,List<SlimAwemeCommentDto> awemeComments,String eventName){
        //开始解析时间
        long start = System.currentTimeMillis();
        //有效评论总量
        final int[] totalComment = {0};
        //语言
        final Map<String,Integer> languageMap = new HashMap<>();
        //活跃的小时
        final int[] activeHours = new int[24];
        //活跃的周内时间
        final int[] activeWeekDays = new int[7];
        //地域
        final Map<String,Integer> regionMap = new HashMap<>();
        //男女比例
        Map<String,Integer> genderMap = new HashMap<>();
        //年龄分布
        Map<String,Integer> ageMap = new HashMap<>();
        //当前时间戳
        long nowTime = System.currentTimeMillis() / 1000;
        //统计时有效的最早时间戳
        long limitTime = nowTime - fansPortraitAppConfig.getStatisticsCommentsLimitDay() * 24 * 3600;
        //周内统计时有效的最早时间戳,因为是周保证是7的倍数
        long limitWeekTime = nowTime - fansPortraitAppConfig.getStatisticsCommentsLimitDay() / 7 * 7 * 24 * 3600;
        //评论用户id
        Set<Long> commentBloggerUid = new HashSet<>();

        //解析
        awemeComments.forEach(awemeComment -> {
            //判断时间是否是在指定区间范围，如果 StatisticsCommentsLimitDay 为 0 则不生效
            try {
                if(awemeComment.getCreateTime() > 0 && (fansPortraitAppConfig.getStatisticsCommentsLimitDay() == 0
                        || awemeComment.getCreateTime() >= limitTime)){
                    //处理时间
                    activeHours[DateTimeUtils.localDateTimeBySecond(awemeComment.getCreateTime()).getHour()]++;
                    if(fansPortraitAppConfig.getStatisticsCommentsLimitDay() == 0
                            || awemeComment.getCreateTime() >= limitWeekTime){
                        activeWeekDays[getWeekDaysByTime(awemeComment.getCreateTime())]++;
                    }
                    //处理语言
                    if(StringUtils.isNotEmpty(awemeComment.getLanguage())){
                        languageMap.put(awemeComment.getLanguage().toLowerCase(),
                                languageMap.getOrDefault(awemeComment.getLanguage().toLowerCase(),0) + 1);
                    }
                    //处理地域
                    if(StringUtils.isNotEmpty(awemeComment.getRegion())){
                        regionMap.put(awemeComment.getRegion().toLowerCase(),
                                regionMap.getOrDefault(awemeComment.getRegion().toLowerCase(),0) + 1);
                    }
                    totalComment[0]++;
                    //添加评论用户id，用来分析男女、性别比例
                    if(awemeComment.getUid() > 0){
                        commentBloggerUid.add(awemeComment.getUid());

                    }
                }
            }catch (Exception e){
                log.info("Aweme comment statistics error. awemeId :{} commentId:{}", awemeId, awemeComment.getCid());
            }
        });

        //解析男女、年龄，返回需要发送调用ai的消息数
        int needSendCount = statisticsGenderAndAge(commentBloggerUid,genderMap,ageMap);
        // 保存数据库
        try{
            BloggerFansPortrait fansPortrait;
            val bloggerFansPortrait = bloggerFansPortraitRepository.findByAwemeId(awemeId);
            if(bloggerFansPortrait.isPresent()){
                fansPortrait = bloggerFansPortrait.get();
            }else{
                val aweme = awemeRepository.findByoAwemeId(awemeId);
                long ctm = aweme.isPresent() ? aweme.get().getOCreateTime() : 0;
                fansPortrait = new BloggerFansPortrait();
                fansPortrait.setUid(uid);
                fansPortrait.setAwemeId(awemeId);
                fansPortrait.setCreateBy(eventName);
                fansPortrait.setAwemeCreateTime(ctm);
                fansPortrait.setCreateTime(System.currentTimeMillis()/1000);
            }
            if(!regionMap.isEmpty() || StringUtils.isEmpty(fansPortrait.getRegionDistribution())){
                fansPortrait.setRegionDistribution(JsonUtil.parseJson(regionMap));
            }
            if(!languageMap.isEmpty() || StringUtils.isEmpty(fansPortrait.getLanguageDistribution())){
                fansPortrait.setLanguageDistribution(JsonUtil.parseJson(languageMap));
            }
            if(!ageMap.isEmpty() || StringUtils.isEmpty(fansPortrait.getAgeDistribution())){
                fansPortrait.setAgeDistribution(JsonUtil.parseJson(ageMap));
            }
            if(!genderMap.isEmpty() || StringUtils.isEmpty(fansPortrait.getGenderDistribution())){
                fansPortrait.setGenderDistribution(JsonUtil.parseJson(genderMap));
            }
            fansPortrait.setActiveDayDistribution(JsonUtil.parseJson(activeHours));
            fansPortrait.setActiveWeekDistribution(JsonUtil.parseJson(activeWeekDays));
            fansPortrait.setCommentsSampleCount(totalComment[0]);
            fansPortrait.setUpdateBy(eventName);
            fansPortrait.setNeedAnalyzeBloggerCount(needSendCount);
            fansPortrait.setUpdateTime(System.currentTimeMillis()/1000);

            bloggerFansPortraitRepository.save(fansPortrait);
        }catch (Exception e){
            log.info("SaveAwemeFansPortraitEvent,save to db error .",e);
        }
        if ((System.currentTimeMillis() - start) / 1000 > fansPortraitAppConfig.getDebugTimeLimit()) {
            log.info("SaveAwemeFansPortraitEvent end awemeId :{} awemeComentSize:{} use :{}s", awemeId, awemeComments.size(), (System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 根据时间戳获取周日期
     * @param time
     * @return
     */
    private int getWeekDaysByTime(long time){
        return DateTimeUtils.localDateTimeBySecond(time).getDayOfWeek().getValue() - 1;
    }

    /**
     * 发送ai查询男女、年龄消息
     * @param uid
     * @param avatarUrl 头像链接
     */
    private void sendMsg(long id,long uid,String avatarUrl){
        //发送消息
        BloggerMessage bloggerMessage = new BloggerMessage();
        bloggerMessage.setId(id);
        bloggerMessage.setUid(uid);
        //存头像
        bloggerMessage.setBloggerType(avatarUrl);
        kafkaTemplateBlogger.send(fansPortraitAppConfig.getBloggerAiExtTopicOnline(),bloggerMessage);
    }

    /**
     * 统计男女、年龄
     * @param uids
     * @param genderMap
     * @param ageMap
     */
    private int statisticsGenderAndAge(Set<Long> uids,Map<String,Integer> genderMap,Map<String,Integer> ageMap){
        //需要发送调用ai的消息数
        int needSendCount = 0;
        if(!uids.isEmpty()){
            Iterator<Long> iterator = uids.iterator();

            List<Long> searchUid = new LinkedList<>();

            while(iterator.hasNext()){
                searchUid.add(iterator.next());
                //查数据库
                if(searchUid.size() == 50){
                    needSendCount = analyzeGenderAndAge(searchUid,genderMap,ageMap,needSendCount);
                    searchUid = new LinkedList<>();
                }
            }
            if(searchUid.size() > 0){
                needSendCount = analyzeGenderAndAge(searchUid,genderMap,ageMap,needSendCount);
            }
        }
        return needSendCount;
    }

    /**
     * 解析男女、年龄比例，未判断过的发送人脸识别男女消息
     * @param searchUid
     * @param genderMap
     * @param ageMap
     * @param needSendCount
     * @return
     */
    private int analyzeGenderAndAge(List<Long> searchUid,Map<String,Integer> genderMap,Map<String,Integer> ageMap,int needSendCount){
        List<SlimBloggerAiExtDto> slimBloggerAiExtDtos;
        SlimBloggerAiExtDto slimBloggerAiExtDto;
        // 年龄区间
        String ageRange;
        // 性别
        String genderDesc;
        try{
            slimBloggerAiExtDtos = bloggerRepository.findBloggersAiExtByInUid(searchUid);
            //分析
            for(int i = 0; i < slimBloggerAiExtDtos.size(); ++i){
                slimBloggerAiExtDto = slimBloggerAiExtDtos.get(i);
                //留下未录入的博主id
                searchUid.remove(slimBloggerAiExtDto.getUid());
                //需要是有效用户
                if(slimBloggerAiExtDto.getTiktokerStatus() == 0){
                    if(slimBloggerAiExtDto.getAiGender() == ConstAi.NOT_ANALYZED){
                        //redis bitmap 判断是否该用户已经发送过
                        try{
                            if(slimBloggerAiExtDto.getId() > 0
                                    && !redisTemplate.opsForValue().getBit("blogger_ai_analyze",slimBloggerAiExtDto.getId())){
                                //发送消息，只发送部分
                                if(!redisTemplate.opsForValue().setBit("blogger_ai_analyze",slimBloggerAiExtDto.getId(),true)
                                        && needSendCount < fansPortraitAppConfig.getBloggerAiExtAwemeCommentsMaxCount()){
                                    sendMsg(slimBloggerAiExtDto.getId(),slimBloggerAiExtDto.getUid(),slimBloggerAiExtDto.getAvatar());
                                }
                            }
                        }catch (Exception e){
                            log.error("Analyze Gender And Age redis bitmap error,msg:{}",e.getMessage());
                        }
                        ++needSendCount;
                    } else if(slimBloggerAiExtDto.getAiGender() == ConstAi.UNKNOWN
                            || slimBloggerAiExtDto.getAiGender() == ConstAi.MALE
                            || slimBloggerAiExtDto.getAiGender() == ConstAi.FEMALE){
                        //处理男女比例
                        switch (slimBloggerAiExtDto.getAiGender()){
                            case 3 : genderDesc = "MALE"; break;
                            case 2 : genderDesc = "FEMALE";break;
                            default:genderDesc = ConstAi.AGE_GENDER_UNKNOWN;
                        }
                        genderMap.put(genderDesc,
                                genderMap.getOrDefault(genderDesc,0) + 1);
                        //处理年龄区间
                        if(slimBloggerAiExtDto.getAiAge() <= ConstAi.UNKNOWN || slimBloggerAiExtDto.getAiAge() > 120){
                            ageRange = ConstAi.AGE_GENDER_UNKNOWN;
                        }else if(slimBloggerAiExtDto.getAiAge() < 18){
                            ageRange = ConstAi.AGE_RANGE_1;
                        }else if(slimBloggerAiExtDto.getAiAge() < 25){
                            ageRange = ConstAi.AGE_RANGE_2;
                        }else if(slimBloggerAiExtDto.getAiAge() < 35){
                            ageRange = ConstAi.AGE_RANGE_3;
                        }else if(slimBloggerAiExtDto.getAiAge() < 45){
                            ageRange = ConstAi.AGE_RANGE_4;
                        }else{
                            ageRange = ConstAi.AGE_RANGE_5;
                        }
                        ageMap.put(ageRange,ageMap.getOrDefault(ageRange,0) + 1);
                    }
                }
            }
            int sendCount = fansPortraitAppConfig.getBloggerAiExtAwemeCommentsMaxCount() - needSendCount;
            needSendCount += searchUid.size();
            //未录入的blogger，先保存空数据再发消息数据
            if(sendCount > 0 && searchUid.size() > 0){
                try {
                    forkJoinPool.submit(() -> searchUid.subList(0,Math.min(sendCount,searchUid.size())).parallelStream().forEach(uid -> {
                        try{
                            if(redisTemplate.opsForValue().setIfAbsent("blogger_ai_ext_"+ uid,"",20, TimeUnit.SECONDS)){
                                //保存数据，获取id
                                Blogger blogger = new Blogger();
                                blogger.setOUid(uid);
                                blogger.setOAvatar300("");
                                blogger.setOAvatar168("");
                                blogger.setOAvatarLarger("");
                                blogger.setOAvatarMedium("");
                                blogger.setOAvatarThumb("");
                                blogger.setODistrict("");
                                blogger.setOSignature("");
                                blogger.setOCoverUrl("");
                                blogger.setOShareInfo("");
                                blogger.setOBirthday("");
                                blogger.setOCountry("");
                                blogger.setOCity("");
                                blogger.setOProvince("");
                                blogger.setOSecUid("");
                                blogger.setOLocation("");
                                blogger.setONickname("");
                                blogger.setORegion("");
                                blogger.setOVideoIcon("");
                                blogger.setOUniqueId("");
                                blogger.setOEnterpriseVerifyReason("");
                                blogger.setWithFusionShopEntry(false);
                                blogger.setCreateBy("BloggerAiAnalyze");
                                blogger.setUpdateBy("BloggerAiAnalyze");
                                blogger = bloggerRepository.save(blogger);
                                redisTemplate.opsForValue().setBit("blogger_ai_analyze",blogger.getId(),true);
                                //发送消息
                                sendMsg(blogger.getId(),uid,null);
                            }
                        }catch(Exception e){
                            log.debug("Analyze Gender And Age insert empty blogger with uid error,msg:{}",e.getMessage());
                        }
                    })).get();
                } catch (Exception ex) {
                    log.error("Analyze Gender And Age thread error.  {}" ,  ex.getMessage());
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            log.info("Analyze Gender And Age error,msg:{}",e.getMessage());
        }
        return needSendCount;
    }

}


